﻿using NsqSharp;
using System;
using System.IO;
using System.Runtime.Serialization.Formatters.Binary;
using System.Text;

namespace log4net.extension.nsq.consumer
{
    class Program
    {
        static string TopicName = "TestLog";
        static string[] nsqAddress => new string[] { "172.16.58.99:9031", "172.16.58.99:9032", "172.16.58.99:9033" };
        static string[] nsqLookupdAddress => new string[] { "172.16.58.99:9012", "172.16.58.99:9011" };

        static void Main(string[] args)
        {
            Console.OutputEncoding = Encoding.UTF8;
            Run_Consumer();
            Console.ReadLine();
        }

        static void Run_Consumer()
        {
            var config = new Config();
            config.DefaultRequeueDelay = TimeSpan.Zero;
            config.MaxBackoffDuration = TimeSpan.FromMilliseconds(50);
            config.MaxInFlight = 6;//建议 nodes*2

            var q = new Consumer(TopicName, "ch2", config);
            q.AddHandler(new MyTestHandler { q = q });

            q.ConnectToNsqd(nsqAddress);
            q.ConnectToNsqLookupd(nsqLookupdAddress);
            var stats = q.GetStats();
            Console.WriteLine("connect to nsq:" + stats.Connections);
        }
    }
    public class MyTestHandler : IHandler
    {
        public Consumer q { get; set; }
        public int messagesSent { get; set; }
        public int messagesReceived { get; set; }
        public int messagesFailed { get; set; }

        public void LogFailedMessage(IMessage message)
        {
            messagesFailed++;
            q.StopAsync();
        }

        public void HandleMessage(IMessage message)
        {
            if (message == null) return;
            try
            {
                var log = Deserialize(message.Body);
                Console.WriteLine(log.Message);
            }
            catch (Exception ex)
            {
                messagesReceived++;
                Console.WriteLine(ex.Message + ex.StackTrace);
            }

            messagesReceived++;
        }

        private Log4netLogEntity Deserialize(byte[] data)
        {
            if (data == null)
                return null;

            using (MemoryStream ms = new MemoryStream(data))
            {
                ms.Position = 0;
                BinaryFormatter formatter = new BinaryFormatter();
                return formatter.Deserialize(ms) as Log4netLogEntity;
            }
        }
    }
}
